package com.freestyle.jmssample.rabbitmq.service;

import com.freestyle.jms.common.AckRabbitTemplate;
import com.freestyle.jmssample.api.SMSConstants;
import com.freestyle.jmssample.api.SendSMSService;
import com.freestyle.jmssample.api.pojo.SMSBean;
import com.freestyle.jmssample.rabbitmq.SMSConfig;
import com.freestyle.jmssample.rabbitmq.SMSUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * Created by rocklee on 2022/1/5 15:59
 */

@Service
@Slf4j
public class SendSMSServiceImp implements SendSMSService {
  @Resource(name="rabbitTemplate")
  private AmqpTemplate amqpTemplate;
  @Resource
  private AckRabbitTemplate ackRabbitTemplate;
  @Resource
  private SMSConfig smsConfig;
  @Override
  public void sendSMS(SMSBean sms) {
    amqpTemplate.convertAndSend(SMSConstants.SMS_EXCHANGE_DIRECT,SMSConstants.SMS_KEY_DIRECT,sms);
  }

  //发送消息返回消费端的回应
  @Override
  public <T> T sendSMSandReceive(SMSBean sms) throws Exception{
    Message message=ackRabbitTemplate.getMessageConverter().toMessage(sms,null);
    CorrelationData correlationData= new CorrelationData(message.getMessageProperties().getMessageId());
    smsConfig.saveSMS(sms,correlationData.getId());
    return (T)(ackRabbitTemplate.convertSendAndReceive(SMSConstants.SMS_EXCHANGE_DIRECT,SMSConstants.SMS_KEY_DIRECT4,message,correlationData));
  }

  //发送消息并需要对方确认
  @Override
  public void sendSMSWithACK(SMSBean sms,String... routerKey) {
   // amqpTemplate.setMandatory(true);
    ackRabbitTemplate.invoke(c->{
      Message message=ackRabbitTemplate.getMessageConverter().toMessage(sms,null);
      CorrelationData correlationData= new CorrelationData(message.getMessageProperties().getMessageId());
      smsConfig.saveSMS(sms,correlationData.getId());
      //要CorrelationDataID与MessageID一致，为了方便confirm/return message回调时拿回消息体。
      ackRabbitTemplate.convertAndSend(SMSConstants.SMS_EXCHANGE_DIRECT,
              routerKey.length==0?SMSConstants.SMS_KEY_DIRECT3:routerKey[0],
              message,correlationData);
      return true;
      /*if (ackRabbitTemplate.waitForConfirms(3000)){//这个确认
        log.info("===ACK sent successfully.===");
        return true;
      }
      else{
        log.info("===ACK sent failed.===");
      }
      return false;*/
    });
  }


  //主题订阅型的发送(发送端要准确的routerkey，订阅方要用通配符)
  @Override
  public void topicSendSMS(SMSBean sms) {
    /*SMSBean sms= SMSUtil.genSMS();
    sms.setSubject("这是发布的消息，欢迎订阅");*/
    Message message=ackRabbitTemplate.getMessageConverter().toMessage(sms,null);
    CorrelationData correlationData= new CorrelationData(message.getMessageProperties().getMessageId());
    smsConfig.saveSMS(sms,correlationData.getId());
    amqpTemplate.convertAndSend(SMSConstants.SMS_EXCHANGE_TOPIC,SMSConstants.SMS_KEY_DIRECT,message);
  }

  @Override
  public void fanOutSendSMS(SMSBean sms) {
    Message message=ackRabbitTemplate.getMessageConverter().toMessage(sms,null);
    CorrelationData correlationData= new CorrelationData(message.getMessageProperties().getMessageId());
    smsConfig.saveSMS(sms,correlationData.getId());
    amqpTemplate.convertAndSend(SMSConstants.SMS_EXCHANGE_FANOUT,"",message);
  }
}
